package com.udf.flink.scala.apitest.watermark

import java.time.Duration

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object LateElement {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
//    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val stream = env.socketTextStream("localhost", 9000, '\n')
    val s = stream
      .map(line => {
        val arr = line.split(",")
        (arr(0), arr(1).toLong * 1000)
      })  //设值获取时间的方式 ，水位线的延迟为5秒钟
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner(new SerializableTimestampAssigner[(String ,Long)]{
        override def extractTimestamp(element: (String, Long), recordTimestamp: Long): Long = element._2
      }) )
//      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long)](Time.seconds(60)) {
//        override def extractTimestamp(element: (String, Long)): Long = element._2
//      })
      .process(new MyLateProcess)

    s.getSideOutput(new OutputTag[String]("late")).print()
    s.print()
    env.execute()
  }
  class MyLateProcess extends ProcessFunction[(String, Long), (String, Long)] {
    val late = new OutputTag[String]("late")
    override def processElement(value: (String, Long),
                                ctx: ProcessFunction[(String, Long), (String, Long)]#Context,
                                out: Collector[(String, Long)]): Unit = {
      println("currentWatermark:"+ctx.timerService().currentWatermark(),ctx.timestamp())
      if (value._2 < ctx.timerService().currentWatermark()) {
        //将低于水位线的迟到数据输出到侧输出流
        ctx.output(late, "这个元素迟到了！"+value._1)
      } else {
        out.collect(value)
      }
    }
  }
//  class MyLateProcess2 extends WatermarkStrategy[(String, Long)] {
//
//  }

}

/*

class MyWaterMark extends AssignerWithPeriodicWatermarks[EventObj] {

  val maxOutOfOrderness = 10000L // 3.0 seconds
  var currentMaxTimestamp = 0L

  val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

  /**
    * 用于生成新的水位线，新的水位线只有大于当前水位线才是有效的
    *
    * 通过生成水印的间隔（每n毫秒）定义 ExecutionConfig.setAutoWatermarkInterval(...)。
    * getCurrentWatermark()每次调用分配器的方法，如果返回的水印非空并且大于先前的水印，则将发出新的水印。
    *
    * @return
    */
  override def getCurrentWatermark: Watermark = {
    new Watermark(this.currentMaxTimestamp - this.maxOutOfOrderness)
  }

  /**
    * 用于从消息中提取事件时间
    *
    * @param element                  EventObj
    * @param previousElementTimestamp Long
    * @return
    */
  override def extractTimestamp(element: EventObj, previousElementTimestamp: Long): Long = {

    currentMaxTimestamp = Math.max(element.timestamp, currentMaxTimestamp)

    val id = Thread.currentThread().getId
    println("currentThreadId:" + id + ",key:" + element.name + ",eventTime:[" + element.datetime + "],currentMaxTimestamp:[" + sdf.format(currentMaxTimestamp) + "],watermark:[" + sdf.format(getCurrentWatermark().getTimestamp) + "]")

    element.timestamp
  }
}
 */